package com.gagakuai.bpm.service.camunda.impl;

/*
 * @program: bpm_server
 * @description: redis延时队列服务
 * @author: houhong
 * @create: 2022-05-10 17:18
 */

import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.TimeUnit;

/*
 * redis延迟队列服务
 */
@Slf4j
@Component
public class RedisDelayJobServiceImpl {

    @Autowired
    private RedissonClient redissonClient;

    /*
     * 添加到消息队列
     *
     * @param value     添加到队列的对象
     * @param delay     延迟时间
     * @param timeUnit  时间单位
     * @param queueName 队列名称
     * @param <T>
     */
    public <T> void addQueue(T value, long delay, TimeUnit timeUnit, String queueName) {
        try {
            log.info("添加到延时队列【{}】【{}】【{}】", value, delay, queueName);
            RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueName);
            RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
            delayedQueue.offer(value, delay, timeUnit);
        } catch (Exception e) {
            log.error("添加到延时队列失败：" + e.getMessage(), e);
            throw new RuntimeException("添加到延时队列失败");
        }
    }


    /*
     * 取值
     */
    public <T> T take(String queueName) throws InterruptedException {
        RBlockingDeque<Map> blockingDeque = redissonClient.getBlockingDeque(queueName);
        T value = (T) blockingDeque.take();
        return value;
    }
}
